package org.itboys.redis;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.itboys.commons.utils.common.ClassUtils;
import org.itboys.commons.utils.string.CommonStringUtils;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

/**
 * Redis 基于 key-value 读写 的client 封装
 * 
 * @author ChenJunhui
 */
public class RedisCacheClient {

	private static Logger logger = LoggerFactory.getLogger(RedisCacheClient.class);

	private RedisConnect redisConnect;

	public <K, T> T get(K key, Class<T> clazz) {
		Jedis jedis = null;
		try {
			jedis = redisConnect.getJedis();
			byte[] value = jedis.get(CommonStringUtils.getBytes(key.toString()));
			if(ArrayUtils.isEmpty(value)) {
				return null;
			}
			return JSON.parseObject(value, clazz);
		} catch (Exception e) {
			logger.error("RedisObjectClient.set error ,key=" + key, e);
			return null;
		} finally {
			redisConnect.returnResource(jedis);
		}
	}
	
	/**
	 * 
	 * @param keys
	 * @param clazz
	 * @return
	 */
	public <T> Map<String, T> get(List<String> keys, Class<T> clazz) {
		if (CollectionUtils.isEmpty(keys)) {
			return Collections.emptyMap();
		}
		for(String key : keys) {
			Preconditions.checkArgument(StringUtils.isNoneBlank(key), "cache key can not be null");
		}
		Map<String, T> result = Maps.newHashMapWithExpectedSize(keys.size());
		Jedis jedis = null;
		try {
			jedis = redisConnect.getJedis();
			Pipeline pipeline = jedis.pipelined();
			List<Response<byte[]>> cacheValues = Lists.newArrayListWithCapacity(keys.size());
			for(String key : keys) {
				Response<byte[]> response = pipeline.get(CommonStringUtils.getBytes(key));
				cacheValues.add(response);
			}
			pipeline.sync();
			for(int i = 0, size = keys.size(); i < size ; i++) {
				String key = keys.get(i);
				Response<byte[]> response = cacheValues.get(i);
				if(response != null) {
					byte[] value = response.get();
					if(ArrayUtils.isNotEmpty(value)) {
						T t = JSON.parseObject(value, clazz);
						result.put(key, t);
					}
				}
			}
			return result;
		} catch (Exception e) {
			logger.error("RedisCacheClient.get error ,key=" + keys, e);
			return Collections.emptyMap();
		} finally {
			redisConnect.returnResource(jedis);
		}
	}

	/**
	 * 往 redis 里放对象
	 * 
	 * @param key
	 *            缓存key
	 * @param object
	 *            缓存对象值
	 * @param expireTime
	 *            超时时间 if <=0 then 忽略超时时间 单位秒
	 * @return
	 */
	public <T> boolean set(String key, T object, int expireTime) {
		Preconditions.checkArgument(StringUtils.isNoneBlank(key)
				&& object != null, "key and value can not be null");
		Jedis jedis = null;
		try {
			jedis = redisConnect.getJedis();
			byte[] value = toJson(object);
			byte[] keyBytes = CommonStringUtils.getBytes(key.toString());
			String result = jedis.set(keyBytes, value);
			boolean success = RedisConstants.SUCCESS_FLAG.equals(result);
			if (success && expireTime > 0) {
				jedis.expire(keyBytes, expireTime);
				return true;
			}
			return success;
		} catch (Exception e) {
			logger.error("RedisCacheClient.set error ,key=" + key, e);
			return false;
		} finally {
			redisConnect.returnResource(jedis);
		}
	}

	public <K, T> void batchSet(Map<K, T> map, int expireTime) {
		if (map == null || map.isEmpty()) {
			return;
		}
		Jedis jedis = null;
		try {
			jedis = redisConnect.getJedis();
			Pipeline pipeline = jedis.pipelined();
			for (K key : map.keySet()) {
				T object = map.get(key);
				if (object == null) {
					continue;
				}
				byte[] value = toJson(object);
				if(value == null) {
					logger.warn("RedisCacheClient.batchSet error ,key={} , value is null", key.toString());
					continue;
				}
				byte[] keyBytes = CommonStringUtils.getBytes(key.toString());
				pipeline.set(keyBytes, value);
				// 设置失效时间
				if (expireTime > 0) {
                    pipeline.expire(keyBytes, expireTime);
                }
			}
			// 批量插入发送一个请求
			pipeline.sync();
		} catch (Exception e) {
			logger.error("RedisCacheClient.batchSet error ,map=" + map, e);
		} finally {
			redisConnect.returnResource(jedis);
		}
	}

	public void del(String... keys) {
		if (keys == null || keys.length == 0) {
			return ;
		}
		Jedis jedis = null;
		try {
			jedis = redisConnect.getJedis();
			Pipeline pipeline = jedis.pipelined();
			for (String key : keys) {
				if(StringUtils.isBlank(key)) {
					continue;
				}
				pipeline.del(key);
			}
			pipeline.sync();
		} catch (Exception e) {
			logger.error("RedisCacheClient.del error ,key=" + Arrays.toString(keys),e);
		} finally {
			redisConnect.returnResource(jedis);
		}
	}

	/**
	 * 加1 并返回加1后的结果
	 * @param key
	 * @return
     */
	public Long incr(String key){
		if (StringUtils.isBlank(key)) {
			return null;
		}
		Jedis jedis = null;
		try{
			jedis = redisConnect.getJedis();
			return jedis.incr(key);
		}catch (Exception e){
			logger.error("RedisCacheClient.incr error, key=" + key,e);
			return null;
		}finally {
			redisConnect.returnResource(jedis);
		}
	}

	public void setRedisConnect(RedisConnect redisConnect) {
		this.redisConnect = redisConnect;
	}

	/**
	 * 判断一个对象是否是基本类型
	 * 
	 * @param obj
	 * @return
	 */
	private boolean isBasedObject(Object obj) {
		if(obj == null) {
			return false;
		}
		return ClassUtils.isSimpleType(obj.getClass());
	}
	
	private byte[] toJson(Object obj) {
		if(obj == null) {
			return null;
		}
        if (isBasedObject(obj)) {
            return CommonStringUtils.getBytes(obj.toString());
        }
        String json = JSON.toJSONString(obj);
        return CommonStringUtils.getBytes(json);
    }
}
